package com.tdy.cdc.app.cdr;

import com.alibaba.fastjson.JSONObject;
import com.tdy.cdc.app.BaseApp;
import com.tdy.cdc.util.DateFormatUtil;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.JDBCUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;

/**
 * @Author: NanHuang
 * @Date: 2023/3/20
 * @Description: 手动获取Kafka中的Topic信息，将数据写入到MySQL中对应的表中
 */
public class CdrToMySQL extends BaseApp {
    public static void main(String[] args) {
        new CdrToMySQL().init(
                6001,
                "CdrToMySQL",
                2
        );
    }

    @Override
    public void invoke(StreamExecutionEnvironment env) {
        // 1 获取所有Topic
        ArrayList<String> topics = readAllTopics();
        // 2 消费数据
        DataStreamSource<String> ds = super.readKafka(env, topics);
        // 3 过滤“AE“数据
        SingleOutputStreamOperator<JSONObject> resultDs = filterData(ds);
        // 4 将数据写入MySQL
        resultDs.addSink(FlinkSinkUtil.getMySQLSink());
    }

    private SingleOutputStreamOperator<JSONObject> filterData(DataStreamSource<String> ds) {
        return ds.map(JSONObject::parseObject)
                .filter(o -> {
                    if (o.containsKey("status")) {
                        return "AA".equals(o.getString("status"));
                    } else {
                        return true;
                    }
                });
    }

    private ArrayList<String> readAllTopics() {
        try {
            Connection conn = JDBCUtil.getConfigConnection();
            String sql = "select target_table from cdr_config_test_copy1";
            PreparedStatement ps = conn.prepareStatement(sql);
            ResultSet resultSet = ps.executeQuery();
            ArrayList<String> result = new ArrayList<>();
            while (resultSet.next()){
                String topic = resultSet.getString(1);
                // 先去掉不存在的Topic
                if (!"cdr_hdsd00_18_02".equals(topic) && !"cdr_hdsd00_18_14".equals(topic) && !"cdr_hdsd00_18_15".equals(topic) && !"cdr_hdsd00_06_02".equals(topic) &&
                !"cdr_hdsd00_08_05".equals(topic) && !"cdr_hdsd00_09_03".equals(topic) && !"cdr_hdsd00_13_02".equals(topic) && !"cdr_hdsd00_14_14".equals(topic) &&
                !"cdr_hdsd00_14_01".equals(topic)){
                    result.add(topic);
                }
            }
            ps.close();
            conn.close();
            return result;
        } catch (SQLException e) {
            throw new RuntimeException("连接MySQL中DWD配置表出错！");
        }
    }
}
